#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | date_add(1) | cst_ds }}'
name='中心滞留代理区汇总'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "代理区code\t代理区名称\t超48小时\t超72小时\t超96小时\t超120小时\t超144小时\t超168小时\t总票件数\t筛单票数-疫情件问题件票数\t筛单票数-非疫情件问题件票数\t筛单票数-未录入问题件票数\t非筛单区域-疫情件问题件票数\t非筛单区域-非疫情件问题件票数\t非筛单区域-未录入问题件票数\t">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 10 --executor-memory 8g --executor-cores 4  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
select
 site_agent_code
,max(site_agent_name) as site_agent_name
,sum(cnt_48h)  as cnt_48h
,sum(cnt_72h)  as cnt_72h
,sum(cnt_96h)  as cnt_96h
,sum(cnt_120h) as cnt_120h
,sum(cnt_144h) as cnt_144h
,sum(cnt_168h) as cnt_168h
,sum(cnt) as cnt
,sum(intercept_yiqing_cnt) intercept_yiqing_cnt
,sum(intercept_other_cnt  ) as  intercept_other_cnt
,sum(intercept_not_cnt    ) as  intercept_not_cnt
,sum(notin_yiqing_cnt     ) as  notin_yiqing_cnt
,sum(notin_other_cnt      ) as  notin_other_cnt
,sum(notin_not_cnt        ) as  notin_not_cnt
from (
    select
    scan_date
    ,site_agent_code
    ,site_agent_name
    ,scan_site_code
    ,scan_site
    ,cnt
    ,hour_type
    ,case    when hour_type='48h'  then cnt else 0 end as cnt_48h
    ,case    when hour_type='72h'  then cnt else 0 end as cnt_72h
    ,case    when hour_type='96h'  then cnt else 0 end as cnt_96h
    ,case    when hour_type='120h' then cnt else 0 end as cnt_120h
    ,case    when hour_type='144h' then cnt else 0 end as cnt_144h
    ,case    when hour_type='168h' then cnt else 0 end as cnt_168h
    ,intercept_cnt
    ,intercept_yiqing_cnt
    ,intercept_other_cnt
    ,intercept_not_cnt
    ,notin_yiqing_cnt
    ,notin_other_cnt
    ,notin_not_cnt
    from
    (
        select
        date(scan_time) scan_date
        ,site_agent_code
        ,max(site_agent_name) as site_agent_name
        ,scan_site_code
        ,max(scan_site) as scan_site
        ,hour_type
        ,count(1) as cnt
        ,sum(is_intercepte) as intercept_cnt  --筛单票数-总票数
        ,sum(if(is_intercepte=1 and is_yiqing_problem=1,1,0)) as intercept_yiqing_cnt                                       --筛单疫情票数
        ,sum(if(is_intercepte=1 and is_yiqing_problem=0 and second_level_type_code is not null,1,0)) as intercept_other_cnt --筛单非疫情票数
        ,sum(if(is_intercepte=1 and is_yiqing_problem=0 and second_level_type_code is  null,1,0)) as intercept_not_cnt      --筛单未录入问题件票数
        ,sum(if(is_intercepte!=1 and is_yiqing_problem=1,1,0)) as notin_yiqing_cnt                                          --非筛单疫情票数
        ,sum(if(is_intercepte!=1 and is_yiqing_problem=0 and second_level_type_code is not null,1,0)) as notin_other_cnt --非筛单非疫情票数
        ,sum(if(is_intercepte!=1 and is_yiqing_problem=0 and second_level_type_code is null,1,0)) as notin_not_cnt      --非筛单未录入问题件票数
        from bidefault.center_yiqing_tmp
        where dt between date_add('{{ execution_date | cst_ds }}',-8) and date_add('{{ execution_date | cst_ds }}',-1)
        and hour_type!='小于48h'
        group by date(scan_time),site_agent_code,scan_site_code,hour_type
    )t
)tt group by site_agent_code;
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt